package externalSort;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

public class mergeConsumer implements Runnable {
    private final BlockingQueue<List<Integer>> inputQueue1;
    private final BlockingQueue<List<Integer>> inputQueue2;
    private final BlockingQueue<List<Integer>> outputQueue;
    private final String outputFile;
    private boolean isFinished = false;

    public mergeConsumer(BlockingQueue<List<Integer>> inputQueue1,
                         BlockingQueue<List<Integer>> inputQueue2,
                         BlockingQueue<List<Integer>> outputQueue,
                         String outputFile) {
        this.inputQueue1 = inputQueue1;
        this.inputQueue2 = inputQueue2;
        this.outputQueue = outputQueue;
        this.outputFile = outputFile;
    }

    @Override
    public void run() {
        try (BufferedWriter writer = new BufferedWriter(new FileWriter(outputFile))) {
            List<Integer> buffer1 = new ArrayList<>();
            List<Integer> buffer2 = new ArrayList<>();
            int index1 = 0, index2 = 0;

            boolean receivedEnd1 = false;
            boolean receivedEnd2 = false;

            while (!receivedEnd1 || !receivedEnd2 || !buffer1.isEmpty() || !buffer2.isEmpty()) {
                // 如果缓冲区空了，从队列获取新数据
                if (buffer1.isEmpty() && !receivedEnd1) {
                    List<Integer> newBuffer = inputQueue1.take();
//                    System.out.println("Consumer "+outputFile+" 从缓存区1中获得了一个Buffer");
                    if (newBuffer.isEmpty()) {//达到了哨兵节点
                        receivedEnd1 = true;
                    } else {
                        buffer1 = newBuffer;
                        index1 = 0;
                    }
                }
                // 同上
                if (buffer2.isEmpty() && !receivedEnd2) {
                    List<Integer> newBuffer = inputQueue2.take();
//                    System.out.println("Consumer "+outputFile+" 从缓存区2中获得了一个Buffer");
                    if (newBuffer.isEmpty()) {
                        receivedEnd2 = true;
                    } else {
                        buffer2 = newBuffer;
                        index2 = 0;
                    }
                }

//                System.out.println(outputFile+" buffer1: "+buffer1.size());
//                for(int num:buffer1){
//                    System.out.print(num+" ");
//                }
//                System.out.print("\n");
//                System.out.println(outputFile+" buffer2: "+buffer2.size());
//                for(int num:buffer2){
//                    System.out.print(num+" ");
//                }
//                System.out.print("\n");


                //Bug在这里！！！
                //如果一个buffer为空的情况下，会导致index<size，会死循环
                if(receivedEnd1||receivedEnd2){
                    break;
                }

                // 执行归并
                while ((index1 < buffer1.size() || !buffer1.isEmpty()) &&
                        (index2 < buffer2.size() || !buffer2.isEmpty())) {

                    if (index1 >= buffer1.size() && !buffer1.isEmpty()) {
                        // buffer1耗尽，重新获取
//                        System.out.println(outputFile+" buffer1耗尽");
                        break;
                    }
                    if (index2 >= buffer2.size() && !buffer2.isEmpty()) {
                        // buffer2耗尽，重新获取
//                        System.out.println(outputFile+" buffer2耗尽");
                        break;
                    }

                    if (index1 < buffer1.size() && index2 < buffer2.size()) {
                        if (buffer1.get(index1) <= buffer2.get(index2)) {
                            writer.write(buffer1.get(index1) + "\n");
//                            System.out.println("写入"+buffer1.get(index1)+" "+index1++);
                            index1++;
                        } else {
                            writer.write(buffer2.get(index2) + "\n");
//                            System.out.println("写入"+buffer2.get(index2)+" "+index2++);
                            index2++;
                        }
                    } else if (index1 < buffer1.size()) {
                        writer.write(buffer1.get(index1) + "\n");
//                        System.out.println("写入"+buffer1.get(index1)+" "+index1++);
                        index1++;
                    } else if (index2 < buffer2.size()) {
                        writer.write(buffer2.get(index2) + "\n");
//                        System.out.println("写入"+buffer2.get(index2)+" "+index2++);
                        index2++;
                    }
                }

                // 检查缓冲区是否耗尽
                if (index1 >= buffer1.size() && !buffer1.isEmpty()) {
                    buffer1.clear();
                    index1 = 0;
                }
                if (index2 >= buffer2.size() && !buffer2.isEmpty()) {
                    buffer2.clear();
                    index2 = 0;
                }
            }
            //判断是否剩余
            while(index1<buffer1.size()){
                writer.write(buffer1.get(index1) + "\n");
                index1++;
            }
            while(index2<buffer2.size()){
                writer.write(buffer2.get(index2) + "\n");
                index2++;
            }

            // 发送结束信号到输出队列（如果需要）
            outputQueue.put(new ArrayList<>());
            isFinished = true;

        } catch (IOException | InterruptedException e) {
            System.out.println("MergeConsumer错误: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }

    public boolean isFinished() {
        return isFinished;
    }
}